package com.atguigu.flink.function;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * Created by Smexy on 2023/2/24
 *
 *  { userId:100，behavior:"uninstall"，"channel":"ZTE"，timestamp:xxx  }
 *
 *   behavior: "install"，"uninstall"，"download"，"update"
 *  channel:"ZTE"，"APPLE"，"HUAWEI"，"MI"，"VIVO"，"OPPO"
 *
 *  SourceFunction: 是一个非并行运行的Source，只能有一个Task产生数据！
 *
 *  ParallelSourceFunction： 是可以并行运行的Source
 */
//public class MySourceFunction implements SourceFunction<String>
public class MySourceFunction implements ParallelSourceFunction<String>
{
    boolean ifContinue = true;
    // 负责产生数据
    @Override
    public void run(SourceContext<String> ctx) throws Exception {

        String [] behavior = {"install","uninstall","download","update" };
        String [] channel = {"ZTE","APPLE","HUAWEI","MI","VIVO","OPPO" };

        //源源不断地产生数据
        while (ifContinue){

            JSONObject jsonObject = new JSONObject();
            jsonObject.put("userId", RandomUtils.nextInt(1,10000));
            jsonObject.put("behavior",behavior[RandomUtils.nextInt(0,behavior.length)]);
            jsonObject.put("channel",channel[RandomUtils.nextInt(0,channel.length)]);
            jsonObject.put("timestamp",System.currentTimeMillis());

            //向下游输出  看到了 xxxContext，就代表运行环境，从中获取很多工具。
            ctx.collect(jsonObject.toJSONString());

            //Thread.sleep(500);

        }

    }

    //停止产生数据
    @Override
    public void cancel() {
            ifContinue = false;
    }
}
